package com.xxl.job.admin.core.thread;

import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
import com.xxl.job.admin.core.cron.CronExpression;
import com.xxl.job.admin.core.model.XxlJobInfo;
import com.xxl.job.admin.core.scheduler.MisfireStrategyEnum;
import com.xxl.job.admin.core.scheduler.ScheduleTypeEnum;
import com.xxl.job.admin.core.trigger.TriggerTypeEnum;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * @author xuxueli 2019-05-21
 */
public class JobScheduleHelper {

  private static Logger logger = LoggerFactory.getLogger(JobScheduleHelper.class);

  private static JobScheduleHelper instance = new JobScheduleHelper();

  public static JobScheduleHelper getInstance() {
    return instance;
  }

  public static final long PRE_READ_MS = 5000;    // pre read

  private Thread scheduleThread;
  private Thread ringThread;
  private volatile boolean scheduleThreadToStop = false;
  private volatile boolean ringThreadToStop = false;
  private volatile static Map<Integer, List<Integer>> ringData = new ConcurrentHashMap<>();

  public void start() {

    // schedule thread
    scheduleThread = new Thread(new Runnable() {
      @Override
      public void run() {

        try {
          TimeUnit.MILLISECONDS.sleep(5000 - System.currentTimeMillis() % 1000);
        } catch (InterruptedException e) {
          if (!scheduleThreadToStop) {
            logger.error(e.getMessage(), e);
          }
        }
        logger.info(">>>>>>>>> init xxl-job admin scheduler success.");

        // pre-read count: treadpool-size * trigger-qps (each trigger cost 50ms, qps = 1000/50 = 20)
        int preReadCount = (XxlJobAdminConfig.getAdminConfig().getTriggerPoolFastMax()
            + XxlJobAdminConfig.getAdminConfig().getTriggerPoolSlowMax()) * 20;

        while (!scheduleThreadToStop) {

          // Scan Job
          long start = System.currentTimeMillis();

          Connection conn = null;
          Boolean connAutoCommit = null;
          PreparedStatement preparedStatement = null;

          boolean preReadSuc = true;
          try {

            conn = XxlJobAdminConfig.getAdminConfig().getDataSource().getConnection();
            connAutoCommit = conn.getAutoCommit();
            conn.setAutoCommit(false);

            preparedStatement = conn.prepareStatement(
                "select * from xxl_job_lock where lock_name = 'schedule_lock' for update");
            preparedStatement.execute();

            // tx start

            // 1、pre read
            long nowTime = System.currentTimeMillis();
            List<XxlJobInfo> scheduleList = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao()
                .scheduleJobQuery(nowTime + PRE_READ_MS, preReadCount);
            if (scheduleList != null && scheduleList.size() > 0) {
              // 2、push time-ring
              for (XxlJobInfo jobInfo : scheduleList) {

                // time-ring jump
                if (nowTime > jobInfo.getTriggerNextTime() + PRE_READ_MS) {
                  // 2.1、trigger-expire > 5s：pass && make next-trigger-time
                  logger.warn(">>>>>>>>>>> xxl-job, schedule misfire, jobId = " + jobInfo.getId());

                  // 1、misfire match
                  MisfireStrategyEnum misfireStrategyEnum = MisfireStrategyEnum.match(
                      jobInfo.getMisfireStrategy(), MisfireStrategyEnum.DO_NOTHING);
                  if (MisfireStrategyEnum.FIRE_ONCE_NOW == misfireStrategyEnum) {
                    // FIRE_ONCE_NOW 》 trigger
                    JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.MISFIRE, -1, null,
                        null, null);
                    logger.debug(
                        ">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId());
                  }

                  // 2、fresh next
                  refreshNextValidTime(jobInfo, new Date());

                } else if (nowTime > jobInfo.getTriggerNextTime()) {
                  // 2.2、trigger-expire < 5s：direct-trigger && make next-trigger-time

                  // 1、trigger
                  JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.CRON, -1, null,
                      null, null);
                  logger.debug(
                      ">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId());

                  // 2、fresh next
                  refreshNextValidTime(jobInfo, new Date());

                  // next-trigger-time in 5s, pre-read again
                  if (jobInfo.getTriggerStatus() == 1
                      && nowTime + PRE_READ_MS > jobInfo.getTriggerNextTime()) {

                    // 1、make ring second
                    int ringSecond = (int) ((jobInfo.getTriggerNextTime() / 1000) % 60);

                    // 2、push time ring
                    pushTimeRing(ringSecond, jobInfo.getId());

                    // 3、fresh next
                    refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));

                  }

                } else {
                  // 2.3、trigger-pre-read：time-ring trigger && make next-trigger-time

                  // 1、make ring second
                  int ringSecond = (int) ((jobInfo.getTriggerNextTime() / 1000) % 60);

                  // 2、push time ring
                  pushTimeRing(ringSecond, jobInfo.getId());

                  // 3、fresh next
                  refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));

                }

              }

              // 3、update trigger info
              for (XxlJobInfo jobInfo : scheduleList) {
                XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleUpdate(jobInfo);
              }

            } else {
              preReadSuc = false;
            }

            // tx stop

          } catch (Exception e) {
            if (!scheduleThreadToStop) {
              logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread error:{}", e);
            }
          } finally {

            // commit
            if (conn != null) {
              try {
                conn.commit();
              } catch (SQLException e) {
                if (!scheduleThreadToStop) {
                  logger.error(e.getMessage(), e);
                }
              }
              try {
                conn.setAutoCommit(connAutoCommit);
              } catch (SQLException e) {
                if (!scheduleThreadToStop) {
                  logger.error(e.getMessage(), e);
                }
              }
              try {
                conn.close();
              } catch (SQLException e) {
                if (!scheduleThreadToStop) {
                  logger.error(e.getMessage(), e);
                }
              }
            }

            // close PreparedStatement
            if (null != preparedStatement) {
              try {
                preparedStatement.close();
              } catch (SQLException e) {
                if (!scheduleThreadToStop) {
                  logger.error(e.getMessage(), e);
                }
              }
            }
          }
          long cost = System.currentTimeMillis() - start;

          // Wait seconds, align second
          if (cost < 1000) {  // scan-overtime, not wait
            try {
              // pre-read period: success > scan each second; fail > skip this period;
              TimeUnit.MILLISECONDS.sleep(
                  (preReadSuc ? 1000 : PRE_READ_MS) - System.currentTimeMillis() % 1000);
            } catch (InterruptedException e) {
              if (!scheduleThreadToStop) {
                logger.error(e.getMessage(), e);
              }
            }
          }

        }

        logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread stop");
      }
    });
    scheduleThread.setDaemon(true);
    scheduleThread.setName("xxl-job, admin JobScheduleHelper#scheduleThread");
    scheduleThread.start();

    // ring thread
    ringThread = new Thread(new Runnable() {
      @Override
      public void run() {

        while (!ringThreadToStop) {

          // align second
          try {
            TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis() % 1000);
          } catch (InterruptedException e) {
            if (!ringThreadToStop) {
              logger.error(e.getMessage(), e);
            }
          }

          try {
            // second data
            List<Integer> ringItemData = new ArrayList<>();
            int nowSecond = Calendar.getInstance()
                .get(Calendar.SECOND);   // 避免处理耗时太长，跨过刻度，向前校验一个刻度；
            for (int i = 0; i < 2; i++) {
              List<Integer> tmpData = ringData.remove((nowSecond + 60 - i) % 60);
              if (tmpData != null) {
                ringItemData.addAll(tmpData);
              }
            }

            // ring trigger
            logger.debug(
                ">>>>>>>>>>> xxl-job, time-ring beat : " + nowSecond + " = " + Arrays.asList(
                    ringItemData));
            if (ringItemData.size() > 0) {
              // do trigger
              for (int jobId : ringItemData) {
                // do trigger
                JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null, null);
              }
              // clear
              ringItemData.clear();
            }
          } catch (Exception e) {
            if (!ringThreadToStop) {
              logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread error:{}", e);
            }
          }
        }
        logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread stop");
      }
    });
    ringThread.setDaemon(true);
    ringThread.setName("xxl-job, admin JobScheduleHelper#ringThread");
    ringThread.start();
  }

  private void refreshNextValidTime(XxlJobInfo jobInfo, Date fromTime) throws Exception {
    Date nextValidTime = generateNextValidTime(jobInfo, fromTime);
    if (nextValidTime != null) {
      jobInfo.setTriggerLastTime(jobInfo.getTriggerNextTime());
      jobInfo.setTriggerNextTime(nextValidTime.getTime());
    } else {
      jobInfo.setTriggerStatus(0);
      jobInfo.setTriggerLastTime(0);
      jobInfo.setTriggerNextTime(0);
      logger.warn(
          ">>>>>>>>>>> xxl-job, refreshNextValidTime fail for job: jobId={}, scheduleType={}, scheduleConf={}",
          jobInfo.getId(), jobInfo.getScheduleType(), jobInfo.getScheduleConf());
    }
  }

  private void pushTimeRing(int ringSecond, int jobId) {
    // push async ring
    List<Integer> ringItemData = ringData.get(ringSecond);
    if (ringItemData == null) {
      ringItemData = new ArrayList<Integer>();
      ringData.put(ringSecond, ringItemData);
    }
    ringItemData.add(jobId);

    logger.debug(
        ">>>>>>>>>>> xxl-job, schedule push time-ring : " + ringSecond + " = " + Arrays.asList(
            ringItemData));
  }

  public void toStop() {

    // 1、stop schedule
    scheduleThreadToStop = true;
    try {
      TimeUnit.SECONDS.sleep(1);  // wait
    } catch (InterruptedException e) {
      logger.error(e.getMessage(), e);
    }
    if (scheduleThread.getState() != Thread.State.TERMINATED) {
      // interrupt and wait
      scheduleThread.interrupt();
      try {
        scheduleThread.join();
      } catch (InterruptedException e) {
        logger.error(e.getMessage(), e);
      }
    }

    // if has ring data
    boolean hasRingData = false;
    if (!ringData.isEmpty()) {
      for (int second : ringData.keySet()) {
        List<Integer> tmpData = ringData.get(second);
        if (tmpData != null && tmpData.size() > 0) {
          hasRingData = true;
          break;
        }
      }
    }
    if (hasRingData) {
      try {
        TimeUnit.SECONDS.sleep(8);
      } catch (InterruptedException e) {
        logger.error(e.getMessage(), e);
      }
    }

    // stop ring (wait job-in-memory stop)
    ringThreadToStop = true;
    try {
      TimeUnit.SECONDS.sleep(1);
    } catch (InterruptedException e) {
      logger.error(e.getMessage(), e);
    }
    if (ringThread.getState() != Thread.State.TERMINATED) {
      // interrupt and wait
      ringThread.interrupt();
      try {
        ringThread.join();
      } catch (InterruptedException e) {
        logger.error(e.getMessage(), e);
      }
    }

    logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper stop");
  }


  // ---------------------- tools ----------------------
  public static Date generateNextValidTime(XxlJobInfo jobInfo, Date fromTime) throws Exception {
    ScheduleTypeEnum scheduleTypeEnum = ScheduleTypeEnum.match(jobInfo.getScheduleType(), null);
    if (ScheduleTypeEnum.CRON == scheduleTypeEnum) {
      Date nextValidTime = new CronExpression(jobInfo.getScheduleConf()).getNextValidTimeAfter(
          fromTime);
      return nextValidTime;
    } else if (ScheduleTypeEnum.FIX_RATE
        == scheduleTypeEnum /*|| ScheduleTypeEnum.FIX_DELAY == scheduleTypeEnum*/) {
      return new Date(fromTime.getTime() + Integer.valueOf(jobInfo.getScheduleConf()) * 1000);
    }
    return null;
  }

}
